package cn.learn.server;

import cn.learn.proto.RequestHeader;
import cn.learn.server.jute.BinaryInputArchive;
import cn.learn.server.persistence.FileTxnSnapLog;
import cn.learn.server.util.RequestPathMetricsCollector;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

public class ZooKeeperServer {
    final Deque<ChangeRecord> outstandingChanges = new ArrayDeque<>();
    // this data structure must be accessed under the outstandingChanges lock
    final Map<String, ChangeRecord> outstandingChangesForPath = new HashMap<String, ChangeRecord>();
    // 内存数据库
    private ZKDatabase zkDb;
    protected ServerCnxnFactory serverCnxnFactory;
    private RequestThrottler requestThrottler;
    protected RequestProcessor firstProcessor;
    private final AtomicLong hzxid = new AtomicLong(0);

    private final RequestPathMetricsCollector requestPathMetricsCollector;

    private FileTxnSnapLog txnLogFactory = null;

    public ZooKeeperServer(FileTxnSnapLog txnLog) {
        this.txnLogFactory = txnLog;
        this.requestPathMetricsCollector = new RequestPathMetricsCollector();
    }

    public void startdata() {
        //check to see if zkDb is not null
        if (zkDb == null) {
            zkDb = new ZKDatabase();
        }
        if (!zkDb.isInitialized()) {
            loadData();
        }
    }

    public void loadData(){
        if (zkDb.isInitialized()) {
//            setZxid(zkDb.getDataTreeLastProcessedZxid());
        } else {
            setZxid(zkDb.loadDataBase());
        }
    }

    public void setZxid(long zxid) {
        hzxid.set(zxid);
    }

    public synchronized void startup() {
//        startupWithServerState(State.RUNNING);
        // 设置3个请求处理器
        setupRequestProcessors();
        // 启动请求节流器
        startRequestThrottler();
    }

    protected void setupRequestProcessors() {
        RequestProcessor finalProcessor = new FinalRequestProcessor(this);
        RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);
        ((SyncRequestProcessor) syncProcessor).start();
        firstProcessor = new PrepRequestProcessor(this, syncProcessor);
        ((PrepRequestProcessor) firstProcessor).start();
    }

    protected void startRequestThrottler() {
        requestThrottler = new RequestThrottler(this);
        requestThrottler.start();

    }

    public void setServerCnxnFactory(ServerCnxnFactory factory) {
        serverCnxnFactory = factory;
    }

    public void processPacket(NettyServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
        InputStream inputStream = new ByteBufferInputStream(incomingBuffer);
        BinaryInputArchive bia = BinaryInputArchive.getArchive(inputStream);
        RequestHeader h = new RequestHeader();
        h.deserialize(bia, "header");

//        int conlen = dataInputStream.readInt();
//        byte[] pathByte = new byte[conlen];
//        dataInputStream.readFully(pathByte);
//        String path = new String(pathByte, StandardCharsets.UTF_8);

//        System.out.printf("xid=%d;type=%d;path=%s\r\n",xid,type,path);

        incomingBuffer = incomingBuffer.slice();
        // 添加到request
        Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), incomingBuffer);
        submitRequest(si);
    }

    private void submitRequest(Request si) {
        while (true){
            if (requestThrottler == null) {
                synchronized (this) {
                    try {
                        wait(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }else {
                break;
            }
        }
        requestThrottler.submitRequest(si);
    }

    public void submitRequestNow(Request si) {
        while (true) {
            if (firstProcessor == null) {
                synchronized (this) {
                    try {
                        wait(1000);
                    } catch (Exception e) {

                    }
                }
            }else {
                break;
            }
        }
        try {
            // 开始执行核心业务
            firstProcessor.processRequest(si);
        }catch (Exception e) {

        }
    }

    public long getZxid() {
        return hzxid.get();
    }

    public long getNextZxid() {
        return hzxid.incrementAndGet();
    }

    public ZKDatabase getZKDatabase() {
        return this.zkDb;
    }

    public void takeSnapshot() throws IOException {
        takeSnapshot(false);
    }

    public void takeSnapshot(boolean syncSnap) throws IOException {
        ConcurrentHashMap<Long, Integer> sessionsWithTimeouts = new ConcurrentHashMap<>();
        txnLogFactory.save(zkDb.getDataTree(), sessionsWithTimeouts, syncSnap);
    }

    public DataTree.ProcessTxnResult processTxn(Request request) {
//        TxnHeader hdr = request.getHdr();
//        DataTree.ProcessTxnResult rc = processTxnInDB(hdr, request.getTxn(), request.getTxnDigest());
        DataTree.ProcessTxnResult rc = new DataTree.ProcessTxnResult();
        return rc;
    }

    public RequestPathMetricsCollector getRequestPathMetricsCollector() {
        return requestPathMetricsCollector;
    }

    static class ChangeRecord {
        long zxid;
        String path;
        int childCount;
        byte[] data;

        ChangeRecord(long zxid, String path, int childCount) {
            this.zxid = zxid;
            this.path = path;
            this.childCount = childCount;
        }
    }

    protected enum State {
        INITIAL,
        RUNNING,
        SHUTDOWN,
        ERROR
    }
}
